{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkConf, SparkContext\n",
    "spark = SparkSession.builder.master('local').appName('pyspark_test').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load the data\n",
    "df0 = spark.read.csv('D:/计算机/数据挖掘/data/q2.csv', header=True, inferSchema=True, encoding='utf-8')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "acceleration_x    0\n",
       "acceleration_y    0\n",
       "acceleration_z    0\n",
       "gyro_x            0\n",
       "gyro_y            0\n",
       "gyro_z            0\n",
       "activity          0\n",
       "dtype: int64"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#whether nan\n",
    "df0.toPandas().isna().sum()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df0.toPandas().isna().values.any()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- acceleration_x: double (nullable = true)\n",
      " |-- acceleration_y: double (nullable = true)\n",
      " |-- acceleration_z: double (nullable = true)\n",
      " |-- gyro_x: double (nullable = true)\n",
      " |-- gyro_y: double (nullable = true)\n",
      " |-- gyro_z: double (nullable = true)\n",
      " |-- activity: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df0.printSchema()\n",
    "#check the type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------------+--------------+-------+-------+-------+--------+\n",
      "|acceleration_x|acceleration_y|acceleration_z| gyro_x| gyro_y| gyro_z|activity|\n",
      "+--------------+--------------+--------------+-------+-------+-------+--------+\n",
      "|         0.265|       -0.7814|       -0.0076| -0.059| 0.0325|-29.296|       0|\n",
      "|        0.6722|       -11.233|       -0.2344|-0.1757| 0.0208| 0.1269|       0|\n",
      "|        0.4399|       -14.817|        0.0722|-0.9105| 0.1063|-24.367|       0|\n",
      "|        0.3031|       -0.8125|        0.0888| 0.1199|-0.4099|-29.336|       0|\n",
      "|        0.4814|       -0.9312|        0.0359| 0.0527| 0.4379| 24.922|       0|\n",
      "|        0.4044|       -0.8056|       -0.0956| 0.6925|-0.2179| 25.750|       0|\n",
      "|         0.632|       -11.290|       -0.2982| 0.0548|-0.1896| 0.4473|       0|\n",
      "|         0.667|       -13.503|        -0.088|-0.8094|-0.7938|-14.348|       0|\n",
      "|        0.2704|       -0.8633|        0.1293|-0.4173|-0.1904|-26.759|       0|\n",
      "|         0.469|       -10.740|        0.0219| 0.0388| 11.491| 16.982|       0|\n",
      "|        0.2985|       -0.7172|       -0.0693| 0.2326| 0.4321| 21.009|       0|\n",
      "|        0.6364|       -10.452|         -0.24| 0.1163|-0.1033| 10.822|       0|\n",
      "|        0.5683|       -12.486|        -0.131|-0.4556|-0.5281|-12.407|       0|\n",
      "|        0.2911|       -0.7748|        0.0163|-0.2345|-0.0148|-25.884|       0|\n",
      "|        0.4477|       -11.574|       -0.0172|-0.1081| 0.4016|   0.67|       0|\n",
      "|        0.2424|       -0.7421|       -0.0549| 0.5714|-0.0506| 21.356|       0|\n",
      "|        0.6028|       -10.966|       -0.3046| 0.1674|-0.5065| 10.156|       0|\n",
      "|        0.4852|       -13.397|       -0.0763|-0.8579| 0.0096|-14.015|       0|\n",
      "|        0.3017|       -0.8366|        0.0718|-0.2701|-0.4678|-27.010|       0|\n",
      "|        0.4082|       -10.859|       -0.0375| 0.0848| -0.105|  0.287|       0|\n",
      "+--------------+--------------+--------------+-------+-------+-------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#set the train data and show the data\n",
    "from pyspark.mllib.linalg import Vectors,Vector\n",
    "from pyspark import SparkContext\n",
    "from pyspark.ml.regression import LinearRegression\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.python.pyspark.shell import spark\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions  import *\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.clustering import KMeans\n",
    "sc=SparkContext.getOrCreate()\n",
    "train_data=sc.textFile(\"D:/计算机/数据挖掘/data/q2.csv\")\n",
    "def GetParts(line):\n",
    "    parts = line.split(',')\n",
    "    return parts[0],parts[1],parts[2],parts[3],parts[4],parts[5],parts[6]\n",
    "header = train_data.first()\n",
    "train_data = train_data.filter(lambda row:row != header)\n",
    "train = train_data.map(lambda line: GetParts(line))\n",
    "df = spark.createDataFrame(train,[\"acceleration_x\",\"acceleration_y\",\"acceleration_z\",\"gyro_x\",\"gyro_y\",\"gyro_z\",\"activity\"])\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "#cast the type of data\n",
    "df = df.withColumn(\"acceleration_x\", df[\"acceleration_x\"].cast(FloatType()))\n",
    "df = df.withColumn(\"acceleration_y\", df[\"acceleration_y\"].cast(FloatType()))\n",
    "df = df.withColumn(\"acceleration_z\", df[\"acceleration_z\"].cast(FloatType()))\n",
    "df = df.withColumn(\"gyro_x\", df[\"gyro_x\"].cast(FloatType()))\n",
    "df = df.withColumn(\"gyro_y\", df[\"gyro_y\"].cast(FloatType()))\n",
    "df = df.withColumn(\"gyro_z\", df[\"gyro_z\"].cast(FloatType()))\n",
    "df = df.withColumn(\"activity\", df[\"activity\"].cast(FloatType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------------------------------------------------------------------------------------------------+-----+\n",
      "|features                                                                                                                    |label|\n",
      "+----------------------------------------------------------------------------------------------------------------------------+-----+\n",
      "|[0.26499998569488525,-0.7814000248908997,-0.007600000128149986,-0.05900000035762787,0.032499998807907104,-29.29599952697754]|0.0  |\n",
      "|[0.6722000241279602,-11.232999801635742,-0.23440000414848328,-0.17569999396800995,0.020800000056624413,0.12690000236034393] |0.0  |\n",
      "|[0.4399000108242035,-14.817000389099121,0.0722000002861023,-0.9104999899864197,0.1062999963760376,-24.367000579833984]      |0.0  |\n",
      "|[0.30309998989105225,-0.8125,0.08879999816417694,0.11990000307559967,-0.4099000096321106,-29.336000442504883]               |0.0  |\n",
      "|[0.4814000129699707,-0.9312000274658203,0.03590000048279762,0.05270000174641609,0.43790000677108765,24.922000885009766]     |0.0  |\n",
      "|[0.4043999910354614,-0.8055999875068665,-0.09560000151395798,0.6924999952316284,-0.21789999306201935,25.75]                 |0.0  |\n",
      "|[0.6320000290870667,-11.289999961853027,-0.29820001125335693,0.05480000004172325,-0.18960000574588776,0.447299987077713]    |0.0  |\n",
      "|[0.6669999957084656,-13.503000259399414,-0.08799999952316284,-0.8094000220298767,-0.7937999963760376,-14.347999572753906]   |0.0  |\n",
      "|[0.2703999876976013,-0.8633000254631042,0.12929999828338623,-0.4172999858856201,-0.19040000438690186,-26.759000778198242]   |0.0  |\n",
      "|[0.4690000116825104,-10.739999771118164,0.021900000050663948,0.03880000114440918,11.491000175476074,16.98200035095215]      |0.0  |\n",
      "|[0.2985000014305115,-0.717199981212616,-0.06930000334978104,0.23260000348091125,0.43209999799728394,21.009000778198242]     |0.0  |\n",
      "|[0.6363999843597412,-10.45199966430664,-0.23999999463558197,0.11630000174045563,-0.10329999774694443,10.821999549865723]    |0.0  |\n",
      "|[0.5683000087738037,-12.486000061035156,-0.13099999725818634,-0.45559999346733093,-0.5281000137329102,-12.406999588012695]  |0.0  |\n",
      "|[0.29109999537467957,-0.7748000025749207,0.016300000250339508,-0.2345000058412552,-0.014800000004470348,-25.884000778198242]|0.0  |\n",
      "|[0.44769999384880066,-11.574000358581543,-0.01720000058412552,-0.10809999704360962,0.4016000032424927,0.6700000166893005]   |0.0  |\n",
      "|[0.24240000545978546,-0.7421000003814697,-0.05490000173449516,0.571399986743927,-0.050599999725818634,21.356000900268555]   |0.0  |\n",
      "|[0.6028000116348267,-10.965999603271484,-0.3046000003814697,0.16740000247955322,-0.5065000057220459,10.156000137329102]     |0.0  |\n",
      "|[0.4851999878883362,-13.397000312805176,-0.0763000026345253,-0.8579000234603882,0.009600000455975533,-14.015000343322754]   |0.0  |\n",
      "|[0.30169999599456787,-0.8366000056266785,0.07180000096559525,-0.2700999975204468,-0.46779999136924744,-27.010000228881836]  |0.0  |\n",
      "|[0.4081999957561493,-10.859000205993652,-0.03750000149011612,0.08479999750852585,-0.10499999672174454,0.28700000047683716]  |0.0  |\n",
      "+----------------------------------------------------------------------------------------------------------------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#assemble the data to Vector\n",
    "assembler = VectorAssembler(inputCols=[\"acceleration_x\",\"acceleration_y\",\"acceleration_z\",\"gyro_x\",\"gyro_y\",\"gyro_z\"],outputCol=\"features\")\n",
    "output = assembler.transform(df)\n",
    "label_features = output.select(\"features\", \"activity\").toDF('features','label')\n",
    "label_features.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.7236355485823154"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# split the train data and test data\n",
    "train_data, test_data = label_features.randomSplit([4.0, 1.0], 100)\n",
    "#train the data\n",
    "lR = LogisticRegression(regParam=0.01)\n",
    "lrModel = blor.fit(train_data)\n",
    "result = blorModel.transform(test_data)\n",
    "\n",
    "# caculate the accuracy\n",
    "result.filter(result.label == result.prediction).count()/result.count()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------------------+-----+\n",
      "|features                                      |label|\n",
      "+----------------------------------------------+-----+\n",
      "|[-1.0602,-0.282,-0.0618,0.8069,-0.9107,1.6153]|1    |\n",
      "+----------------------------------------------+-----+\n",
      "\n",
      "[Row(features=DenseVector([-1.0602, -0.282, -0.0618, 0.8069, -0.9107, 1.6153]), label=1, probability=DenseVector([0.4992, 0.5008]), prediction=1.0)]\n"
     ]
    }
   ],
   "source": [
    "#predict the data\n",
    "df1 = spark.createDataFrame([(-1.0602,-0.282,-0.0618,0.8069,-0.9107,1.6153,1)],[\"acceleration_x\",\"acceleration_y\",\"acceleration_z\",\"gyro_x\",\"gyro_y\",\"gyro_z\",\"activity\"])\n",
    "df1.show()\n",
    "test_assembler = VectorAssembler(inputCols=[\"acceleration_x\",\"acceleration_y\",\"acceleration_z\",\"gyro_x\",\"gyro_y\",\"gyro_z\"],outputCol=\"features\")\n",
    "test_output = test_assembler.transform(df1)\n",
    "test_label_features = test_output.select(\"features\", \"activity\").toDF('features','label')\n",
    "test_label_features.show(truncate=False)\n",
    "\n",
    "\n",
    "# df1 = label_features.head(5)\n",
    "# df1 = spark.createDataFrame(df1)\n",
    "# df1.show()\n",
    "prediction = lrModel.transform(test_label_features)\n",
    "result = prediction.select(\"features\", \"label\", \"probability\",\"prediction\").collect()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
